package com.boot.study.table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors._

object EsOutputTest {
  def main(args: Array[String]): Unit = {
    // 1: 创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    val filePath: String = "D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt"
    tableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("inputTable")
    val sensorTable: Table = tableEnv.from("inputTable")
//    var resultTable: Table = sensorTable
//      .select('id, 'temperature)
//      .filter('id === "Sensor_1")

    // 聚合
    val aggTable: Table = sensorTable
      .groupBy('id) // 根据id分组
      .select('id, 'id.count as 'count)

    // 输出es
    // 查看 http://127.0.0.1:9200/sensor/_search?pretty
    tableEnv.connect(new Elasticsearch()
      .version("6")
      .host("localhost", 9200, "http")
      .index("sensor")
      .documentType("temperature")
    )
      .inUpsertMode()
      .withFormat(new Json())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("count", DataTypes.BIGINT())
      )
      .createTemporaryTable("esOutputTable")
    aggTable.insertInto("esOutputTable")

    env.execute("es output test")
  }
}
